from collections import defaultdict
from datetime import datetime
from pathlib import Path
import re
import numpy as np
import pandas as pd
from pandasql import sqldf
import json
from decimal import Decimal
from app.models.db import DATABASE_URL


class DataHelp:
    def __init__(self):
        self._init_logger()

    def _init_logger(self):
        """初始化日志记录系统"""
        self.logs = []
        self.stats = defaultdict(int)
        self.start_time = datetime.now()

    def generate_analysis_report(self, df):
        # 分析数据,获取报告
        df = df.convert_dtypes()
        # 获取分析概要
        summary = df.describe().to_markdown()
        return df
      # 类型优化减少内存占用

    def extract_bracketed_content(text):
        pattern = r'\[(.*?)\]'
        matches = re.findall(pattern, text)
        return matches

    def read_excel_file(file_path: str) -> pd.DataFrame:
        """
        读取Excel文件并返回DataFrame
        :param file_path: Excel文件路径
        :return: DataFrame
        """
        file_extension = Path(file_path).suffix.lower()
        if file_extension == '.xlsx':
            df = pd.read_excel(file_path, engine='openpyxl')
        elif file_extension == '.xls':
            df = pd.read_excel(file_path, engine='xlrd')
        else:
            raise ValueError("不支持的文件格式")
        return df

    @staticmethod
    def get_table_name(sessionid):
        return 'udb_' + sessionid.replace('-', '')

    @staticmethod
    async def get_dataset_structure(dataset_id: str, session) -> dict:
        """
        获取数据集的结构信息，优先使用DataField表中的字段信息
        
        参数:
            dataset_id (str): 数据集ID
            session: 数据库会话对象
            
        返回:
            dict: 数据集结构信息
        """
        from app.models.db import DataSet, DataField, DataSource
        import json
        from loguru import logger
        from sqlalchemy.future import select
        
        logger.debug(f"获取数据集结构信息: {dataset_id}")
        
        try:
            # 1. 获取数据集信息
            dataset_query = select(DataSet).where(DataSet.id == dataset_id)
            dataset_result = await session.execute(dataset_query)
            dataset = dataset_result.scalar_one_or_none()
            
            if not dataset:
                raise ValueError(f"数据集不存在: {dataset_id}")
            
            # 2. 获取数据源信息
            data_source_info = {}
            if dataset.data_source_id != "00000000-0000-0000-0000-000000000000":
                # 查询数据源信息
                datasource_query = select(DataSource).where(DataSource.id == dataset.data_source_id)
                datasource_result = await session.execute(datasource_query)
                datasource = datasource_result.scalar_one_or_none()
                
                if datasource:
                    data_source_info = {
                        "数据库类型": datasource.type,
                        "数据库描述": datasource.description
                    }
            else:
                data_source_info = {
                        "数据库类型": DataHelp.get_db_type_from_url(DATABASE_URL),
                        "数据库描述": "无"
                }
            
            # 3. 从DataField表获取字段信息
            fields_query = select(DataField).where(
                DataField.dataset_id == dataset_id,
                DataField.visible == "1"
            ).order_by(DataField.ordinal_position)
            
            fields_result = await session.execute(fields_query)
            fields = fields_result.scalars().all()
            
            if not fields:
                return {
                    "error": "未找到字段信息，请先创建字段记录",
                    "table_name": dataset.original_name or "",
                    "description": dataset.description or "",
                    "fields": []
                }
            
            # 4. 构建字段信息列表
            columns_info = []
            numeric_columns = []
            date_columns = []
            text_columns = []
            categorical_columns = []
            boolean_columns = []
            
            for field in fields:
                # 构建字段信息
                col_info = {
                    "名称": field.name,
                    "原始类型": field.original_data_type,
                    "显示名称": field.display_name or field.name,
                    "是否主键": field.is_primary_key,
                    "语义类别": field.semantic_type or "文本",
                    "字段描述": field.description or ""
                }
                columns_info.append(col_info)
                
                # 根据语义类型分类
                semantic_type = field.semantic_type or ""
                data_type = field.data_type.lower() if field.data_type else ""
                
                if semantic_type == "数值" or data_type in ["int", "integer", "bigint", "smallint", "float", "double", "decimal", "numeric"]:
                    numeric_columns.append(field.name)
                elif semantic_type == "日期" or data_type in ["date", "datetime", "timestamp", "time"]:
                    date_columns.append(field.name)
                elif semantic_type == "布尔" or data_type in ["bool", "boolean", "bit"]:
                    boolean_columns.append(field.name)
                elif semantic_type == "分类":
                    categorical_columns.append(field.name)
                else:
                    text_columns.append(field.name)
            
         
            # 6. 构建数据集结构信息
            structure = {
                "表名": dataset.original_name or "",
                "表描述": dataset.description or "",
                "数据集名称": dataset.name,
                "列信息": columns_info,
                "列数量": len(columns_info),
                "数据库类型": data_source_info.get("数据库类型", "未知"),
                "数据库描述": data_source_info.get("数据库描述", "无")
            }
            
            # 7. 添加列类型分类
            structure["列类型分类"] = {
                "数值型列": numeric_columns,
                "日期型列": date_columns,
                "文本型列": text_columns,
                "分类型列": categorical_columns,
                "布尔型列": boolean_columns
            }
            
            # 8. 添加适合大模型的描述信息
            description = (f"数据集 {dataset.name} (表名: {dataset.original_name})，包含 {len(columns_info)} 列。"
                          f"其中数值型列 {len(numeric_columns)} 个，日期型列 {len(date_columns)} 个，"
                          f"文本型列 {len(text_columns)} 个，分类型列 {len(categorical_columns)} 个，"
                          f"布尔型列 {len(boolean_columns)} 个。")
            
            structure["数据描述"] = description
            
            return structure
            
        except Exception as e:
            logger.error(f"获取数据集结构失败: {str(e)}")
            return {"错误": f"获取数据集结构失败: {str(e)}"}

    @staticmethod
    def convert_numpy_types(obj):  # 类型转换工具函数
        if isinstance(obj, np.generic):
            return obj.item()
        elif isinstance(obj, dict):
            return {k: DataHelp.convert_numpy_types(v) for k, v in obj.items()}
        elif isinstance(obj, list):
            return [DataHelp.convert_numpy_types(v) for v in obj]
        # 添加对Decimal类型的支持
        elif isinstance(obj, Decimal):
            return float(obj)
        # 添加对datetime类型的支持
        elif hasattr(obj, 'isoformat') and callable(getattr(obj, 'isoformat')):
            return obj.isoformat()
        # 添加对UUID类型的支持
        elif hasattr(obj, 'hex') and hasattr(obj, 'bytes') and hasattr(obj, 'version'):
            return str(obj)
        # 添加对SQLAlchemy行对象的支持
        elif hasattr(obj, '_mapping'):
            return dict(obj._mapping)
        # 添加对blob/binary数据的支持
        elif isinstance(obj, bytes):
            return obj.hex()
        return obj

    @staticmethod
    def to_json(obj, ensure_ascii=False):
        """
        将对象转换为 JSON 字符串，自动处理 NumPy 类型和其他常见数据库类型
        
        参数:
            obj: 要转换的对象
            ensure_ascii: 是否确保 ASCII 编码，默认为 False
            
        返回:
            str: JSON 字符串
        """
        # 定义一个自定义的JSON编码器
        class DatabaseTypeEncoder(json.JSONEncoder):
            def default(self, o):
                # 处理Decimal类型
                if isinstance(o, Decimal):
                    return float(o)
                # 处理日期时间类型
                elif hasattr(o, 'isoformat') and callable(getattr(o, 'isoformat')):
                    return o.isoformat()
                # 处理UUID类型
                elif hasattr(o, 'hex') and hasattr(o, 'bytes') and hasattr(o, 'version'):
                    return str(o)
                # 处理SQLAlchemy行对象
                elif hasattr(o, '_mapping'):
                    return dict(o._mapping)
                # 处理二进制数据
                elif isinstance(o, bytes):
                    return o.hex()
                # 处理NumPy类型
                elif isinstance(o, np.generic):
                    return o.item()
                # 其他类型使用默认处理
                return super().default(o)
        
        try:
            # 使用自定义编码器直接转换
            return json.dumps(obj, ensure_ascii=ensure_ascii, cls=DatabaseTypeEncoder)
        except:
            # 如果失败，回退到原来的方式
            converted_obj = DataHelp.convert_numpy_types(obj)
            return json.dumps(converted_obj, ensure_ascii=ensure_ascii)

    @staticmethod
    def convert_to_markdown_table(query_result):
        """
        将query_result转换为Markdown表格格式
        :param query_result: 查询结果，列表形式，每个元素是一个字典
        :return: Markdown格式的表格字符串
        """
        if not query_result:
            return "No data found."

        # 获取表头
        headers = query_result[0].keys()
        # 生成表头行
        header_row = "| " + " | ".join(headers) + " |"
        # 生成分隔行
        separator_row = "| " + " | ".join(["---"] * len(headers)) + " |"
        # 生成数据行
        data_rows = []
        for row in query_result:
            data_row = "| " + " | ".join(str(row[header])
                                         for header in headers) + " |"
            data_rows.append(data_row)

        # 拼接成完整的Markdown表格
        markdown_table = "\n".join([header_row, separator_row] + data_rows)
        return markdown_table

    def convert_to_markdown_table1(query_result):
        """
        将query_result转换为带有边框的HTML表格格式
        :param query_result: 查询结果，列表形式，每个元素是一个字典
        :return: 带有边框的HTML表格字符串
        """
        if not query_result:
            return "<p>No data found.</p>"

        # 获取表头
        headers = query_result[0].keys()

        # 生成HTML表格
        html_table = "<table class='markdown-table'>\n"
        # 生成表头行
        html_table += "  <tr>\n"
        for header in headers:
            html_table += f"    <th>{header}</th>\n"
        html_table += "  </tr>\n"
        # 生成数据行
        for row in query_result:
            html_table += "  <tr>\n"
            for header in headers:
                html_table += f"    <td>{row[header]}</td>\n"
            html_table += "  </tr>\n"
        html_table += "</table>"

        return html_table

    @staticmethod
    def add_sequence_column(df: pd.DataFrame) -> pd.DataFrame:
        """
        为DataFrame添加一列序号
        :param df: 输入的DataFrame
        :return: 添加了序号列的DataFrame
        """
        df['序号'] = range(1, len(df) + 1)
        # 将序号列移到第一列
        cols = df.columns.tolist()
        cols = ['序号'] + [col for col in cols if col != '序号']
        df = df[cols]
        return df


    @staticmethod
    def get_dataframe_structure(df, virtual_table_name="data",table_desc=""):
        """
        直接从pandas DataFrame获取结构信息，包括列名、列类型等，用于传递给大模型
        
        参数:
            df (pandas.DataFrame): 要分析的DataFrame
            virtual_table_name (str): 虚拟表名，用于生成SQL提示
            
        返回:
            dict: 包含数据结构信息的字典，格式如下:
            {
                "table_name": "虚拟表名",
                "columns": [
                    {"name": "列名1", "type": "列类型1"},
                    {"name": "列名2", "type": "列类型2"},
                    ...
                ],
                "sample_data": [行1, 行2, ...] (最多5条)
            }
        """
        try:
            # 获取列信息
            columns_info = []
            
            # 分析DataFrame的dtypes
            for col_name, dtype in df.dtypes.items():
            
                # 转换pandas数据类型为SQL类型的近似值
                sql_type = DataHelp._map_pandas_type_to_sql(dtype)
                
                col_info = {
                    "name": col_name,
                    "type": sql_type
                }
                columns_info.append(col_info)
            
            # 获取样本数据（最多5条）并确保日期时间格式正确处理
            sample_data = []
            sample_df = df.head(5)
            for _, row in sample_df.iterrows():
                row_dict = {}
                for col in sample_df.columns:
                    value = row[col]
                    # 处理日期时间类型
                    if pd.isna(value):
                        row_dict[col] = None
                    elif isinstance(value, (pd.Timestamp, datetime)):
                        row_dict[col] = value.strftime('%Y-%m-%d %H:%M:%S')
                    else:
                        row_dict[col] = value
                sample_data.append(row_dict)
            
            # 构建结构信息
            structure = {
                "table_name": virtual_table_name,
                "table_desc": table_desc,
                "columns": columns_info,
                "sample_data": sample_data,
                "row_count": len(df)
            }
            
            # 尝试识别数值型、日期型和文本型列
            numeric_columns = []
            date_columns = []
            text_columns = []
            categorical_columns = []
            boolean_columns = []
            
            # 更智能地识别列类型
            for col_name in df.columns:
                # 获取pandas数据类型的字符串表示
                dtype_str = str(df[col_name].dtype).lower()
                
                # 数值型列识别
                if df[col_name].dtype.kind in 'ifc':  # 整数、浮点数或复数
                    numeric_columns.append(col_name)
                # 日期型列识别
                elif df[col_name].dtype.kind == 'M' or 'datetime' in dtype_str or 'timestamp' in dtype_str:
                    date_columns.append(col_name)
                # 布尔型列识别
                elif df[col_name].dtype.kind == 'b' or df[col_name].dtype == 'bool':
                    boolean_columns.append(col_name)
                # 文本型列识别
                elif df[col_name].dtype.kind == 'O' or 'str' in dtype_str or 'object' in dtype_str:
                    # 检查是否可能是分类列（唯一值较少）
                    unique_ratio = df[col_name].nunique() / len(df) if len(df) > 0 else 0
                    if unique_ratio < 0.1 and df[col_name].nunique() < 20:  # 少于10%的唯一值比例且不超过20个唯一值
                        categorical_columns.append(col_name)
                    else:
                        text_columns.append(col_name)
                # 分类型列识别
                elif 'category' in dtype_str:
                    categorical_columns.append(col_name)
            
            # 添加列类型分类
            structure["column_types"] = {
                "numeric_columns": numeric_columns,
                "date_columns": date_columns,
                "text_columns": text_columns,
                "categorical_columns": categorical_columns,
                "boolean_columns": boolean_columns
            }
            
            # 添加数据统计信息
            stats = {}
            for col in numeric_columns:
                try:
                    stats[col] = {
                        "min": float(df[col].min()) if not pd.isna(df[col].min()) else None,
                        "max": float(df[col].max()) if not pd.isna(df[col].max()) else None,
                        "mean": float(df[col].mean()) if not pd.isna(df[col].mean()) else None,
                        "median": float(df[col].median()) if not pd.isna(df[col].median()) else None
                    }
                except:
                    # 跳过无法计算统计值的列
                    pass
            
            structure["statistics"] = stats
            
            # 添加适合大模型的描述信息
            description = (f"数据包含{len(df)}行, {len(columns_info)}列。其中数值型列{len(numeric_columns)}个, "
                        f"日期型列{len(date_columns)}个, 文本型列{len(text_columns)}个, "
                        f"分类型列{len(categorical_columns)}个, 布尔型列{len(boolean_columns)}个。")
            
            structure["description"] = description
            
            # 添加用于生成SQL的提示
            sql_hints = []
            
            # 添加数值型列的聚合函数提示
            if numeric_columns:
                sql_hints.append(f"数值列 ({', '.join(numeric_columns)}) 可以使用SUM, AVG, MAX, MIN等聚合函数")
            
            # 添加日期型列的日期函数提示
            if date_columns:
                sql_hints.append(f"日期列 ({', '.join(date_columns)}) 可以使用STRFTIME函数进行日期格式化和时间周期提取")
            
            # 添加文本型列的文本函数提示
            if text_columns:
                sql_hints.append(f"文本列 ({', '.join(text_columns)}) 可以使用LIKE操作符进行模糊匹配")
            
            # 添加分类列的提示
            if categorical_columns:
                sql_hints.append(f"分类列 ({', '.join(categorical_columns)}) 适合进行GROUP BY分组和分类统计")
            
            # 添加布尔列的提示
            if boolean_columns:
                sql_hints.append(f"布尔列 ({', '.join(boolean_columns)}) 可用于条件过滤")
            
            structure["sql_hints"] = sql_hints
            
            # 添加常见分析类型建议
            analysis_suggestions = []
            
            # 时间序列分析建议
            if date_columns and numeric_columns:
                analysis_suggestions.append({
                    "type": "时间序列分析",
                    "description": f"可以按照日期列 ({', '.join(date_columns)}) 分析数值列 ({', '.join(numeric_columns[:3])}) 随时间的变化趋势"
                })
            
            # 分类统计分析建议
            if categorical_columns and numeric_columns:
                analysis_suggestions.append({
                    "type": "分类统计分析",
                    "description": f"可以按照分类列 ({', '.join(categorical_columns[:3])}) 对数值列 ({', '.join(numeric_columns[:3])}) 进行分组统计"
                })
            
            # 相关性分析建议
            if len(numeric_columns) >= 2:
                analysis_suggestions.append({
                    "type": "相关性分析",
                    "description": f"可以分析数值列之间的相关关系，如 {', '.join(numeric_columns[:3])} 之间的相关性"
                })
            
            structure["analysis_suggestions"] = analysis_suggestions
            
            return structure
            
        except Exception as e:
            return {"error": f"分析DataFrame结构失败: {str(e)}"}

    @staticmethod
    def _map_pandas_type_to_sql(pandas_dtype):
        """
        将pandas数据类型映射为SQL类型的字符串表示
        
        参数:
            pandas_dtype: pandas的数据类型
            
        返回:
            str: 对应的SQL类型字符串
        """
        dtype_str = str(pandas_dtype).lower()
        
        # 整数类型
        if pandas_dtype.kind == 'i':
            if 'int8' in dtype_str or 'int16' in dtype_str:
                return 'SMALLINT'
            elif 'int32' in dtype_str:
                return 'INTEGER'
            else:
                return 'BIGINT'
        
        # 浮点类型
        elif pandas_dtype.kind == 'f':
            if 'float32' in dtype_str:
                return 'FLOAT'
            else:
                return 'DOUBLE'
        
        # 布尔类型
        elif pandas_dtype.kind == 'b':
            return 'BOOLEAN'
        
        # 日期时间类型
        elif pandas_dtype.kind == 'M' or 'datetime' in dtype_str:
            return 'TIMESTAMP'
        
        # 字符串和对象类型
        elif pandas_dtype.kind == 'O' or 'object' in dtype_str:
            return 'VARCHAR'
        
        # 分类类型
        elif 'category' in dtype_str:
            return 'VARCHAR'
        
        # 默认为TEXT
        return 'TEXT'

    @staticmethod
    def get_data_summary(json_data):
        """
        根据JSON格式的表格数据生成摘要信息，用于传递给大模型
        
        参数:
            json_data (list): JSON格式的表格数据，每个元素是一个表示行的字典
            
        返回:
            str: 格式化的数据摘要信息
        """
        try:
            if not json_data or not isinstance(json_data, list) or len(json_data) == 0:
                return "数据为空或格式不正确"
            
            # 基本信息统计
            row_count = len(json_data)
            first_row = json_data[0]
            all_fields = list(first_row.keys())
            field_count = len(all_fields)
            
            # 字段类型识别
            field_types = {}
            numeric_fields = []
            text_fields = []
            date_fields = []
            boolean_fields = []
            categorical_candidates = {}  # 用于跟踪可能的分类字段及其唯一值
            
            # 首次遍历：初步识别字段类型
            for field in all_fields:
                # 初始化可能的分类字段唯一值集合
                categorical_candidates[field] = set()
                
                # 获取第一个非空值
                first_value = None
                for row in json_data:
                    if field in row and row[field] is not None:
                        first_value = row[field]
                        break
                
                if first_value is None:
                    # 如果所有值都为空，默认为文本型
                    field_types[field] = "text"
                    text_fields.append(field)
                elif isinstance(first_value, (int, float)):
                    # 数值型
                    field_types[field] = "numeric"
                    numeric_fields.append(field)
                elif isinstance(first_value, bool):
                    # 布尔型
                    field_types[field] = "boolean"
                    boolean_fields.append(field)
                elif isinstance(first_value, str):
                    # 尝试识别日期型
                    if re.match(r'\d{4}[-/]\d{1,2}[-/]\d{1,2}', first_value):
                        field_types[field] = "date"
                        date_fields.append(field)
                    else:
                        field_types[field] = "text"
                        text_fields.append(field)
                else:
                    # 其他类型默认为文本
                    field_types[field] = "text"
                    text_fields.append(field)
            
            # 第二次遍历：收集唯一值和统计信息
            stats = {}
            for field in all_fields:
                stats[field] = {
                    "null_count": 0,
                    "values": []
                }
                
                if field_types[field] == "numeric":
                    stats[field].update({
                        "min": float('inf'),
                        "max": float('-inf'),
                        "sum": 0,
                        "valid_count": 0
                    })
            
            for row in json_data:
                for field in all_fields:
                    if field not in row or row[field] is None:
                        stats[field]["null_count"] += 1
                        continue
                    
                    value = row[field]
                    
                    # 收集可能的分类字段的唯一值（最多100个）
                    if field_types[field] in ["text", "date"] and len(categorical_candidates[field]) < 100:
                        categorical_candidates[field].add(str(value))
                    
                    # 统计数值型字段
                    if field_types[field] == "numeric":
                        stats[field]["min"] = min(stats[field]["min"], value)
                        stats[field]["max"] = max(stats[field]["max"], value)
                        stats[field]["sum"] += value
                        stats[field]["valid_count"] += 1
                    
                    # 收集样本值（最多5个）
                    if len(stats[field]["values"]) < 5 and value not in stats[field]["values"]:
                        stats[field]["values"].append(value)
            
            # 识别分类字段（唯一值较少的文本字段）
            categorical_fields = []
            for field, unique_values in categorical_candidates.items():
                if field_types[field] in ["text", "date"]:
                    # 如果唯一值数量小于行数的10%且不超过20个，则认为是分类字段
                    if len(unique_values) <= min(20, row_count * 0.1):
                        categorical_fields.append(field)
                        # 如果原来是文本字段，则从文本字段列表中移除
                        if field in text_fields:
                            text_fields.remove(field)
                        # 如果原来是日期字段，则从日期字段列表中移除
                        elif field in date_fields:
                            date_fields.remove(field)
            
            # 计算数值字段的平均值
            for field in numeric_fields:
                if stats[field]["valid_count"] > 0:
                    stats[field]["avg"] = stats[field]["sum"] / stats[field]["valid_count"]
                else:
                    stats[field]["avg"] = None
            
            # 构建摘要信息
            summary = f"数据共{row_count}行，包含{field_count}个字段，其中："
            
            field_type_summary = []
            if numeric_fields:
                field_type_summary.append(f"数值型字段{len(numeric_fields)}个（{', '.join(numeric_fields)}）")
            if date_fields:
                field_type_summary.append(f"日期型字段{len(date_fields)}个（{', '.join(date_fields)}）")
            if categorical_fields:
                field_type_summary.append(f"分类型字段{len(categorical_fields)}个（{', '.join(categorical_fields)}）")
            if text_fields:
                field_type_summary.append(f"文本型字段{len(text_fields)}个（{', '.join(text_fields)}）")
            if boolean_fields:
                field_type_summary.append(f"布尔型字段{len(boolean_fields)}个（{', '.join(boolean_fields)}）")
            
            summary += "；".join(field_type_summary) + "。\n\n"
            
            # 添加数值字段统计信息
            if numeric_fields:
                summary += "数值字段统计：\n"
                for field in numeric_fields:
                    if stats[field]["valid_count"] > 0:
                        summary += f"- {field}：最小值 {stats[field]['min']:.2f}，最大值 {stats[field]['max']:.2f}，平均值 {stats[field]['avg']:.2f}，非空值数量 {stats[field]['valid_count']}\n"
            
            # 添加分类字段信息
            if categorical_fields:
                summary += "\n分类字段信息：\n"
                for field in categorical_fields:
                    unique_count = len(categorical_candidates[field])
                    values_preview = ", ".join([str(v) for v in list(categorical_candidates[field])[:5]])
                    if len(categorical_candidates[field]) > 5:
                        values_preview += "..."
                    summary += f"- {field}：共{unique_count}个不同值，示例：{values_preview}\n"
            
            # 添加日期字段信息
            if date_fields:
                summary += "\n日期字段信息：\n"
                for field in date_fields:
                    summary += f"- {field}：样本值 {', '.join([str(v) for v in stats[field]['values']])}\n"
            
            # 添加数据完整性信息
            null_rates = []
            for field in all_fields:
                null_rate = stats[field]["null_count"] / row_count * 100
                if null_rate > 5:  # 只关注缺失率超过5%的字段
                    null_rates.append(f"{field} ({null_rate:.1f}%)")
            
            if null_rates:
                summary += f"\n注意：以下字段存在较高比例的缺失值：{', '.join(null_rates)}\n"
            
            return summary
            
        except Exception as e:
            return f"生成数据摘要时出错：{str(e)}"

    @staticmethod
    async def generate_dataset_summary(dataset_id, session,search=""):
        """
        根据数据集ID生成数据摘要信息
        
        参数:
            dataset_id (str): 数据集ID
            session: 数据库会话对象
            
        返回:
            dict: 数据摘要信息
        """
        try:
            from app.services.dataset_service import DatasetService
            from app.datatk.data_clear import EnhancedDataCleaner
            import pandas as pd
            from loguru import logger
            
            # 获取数据集数据
            dataset_data = await DatasetService.get_dataset_data(dataset_id, page=1, page_size=50000, session=session,search=search)
            
            if not dataset_data["success"]:
                return {
                    "错误": f"获取数据集数据失败: {dataset_data.get('message', '未知错误')}"
                }
            
            # 如果返回的数据中包含dataframe字段，直接使用
            if "dataframe" in dataset_data and isinstance(dataset_data["dataframe"], pd.DataFrame):
                df = dataset_data["dataframe"]
            else:
                # 否则，从data字段创建DataFrame
                df = pd.DataFrame(dataset_data["data"])
            
            if df.empty:
                return {
                    "警告": "数据集为空，无法生成摘要信息",
                    "基本信息": {
                        "总行数": 0,
                        "总列数": len(dataset_data.get("columns", [])),
                    }
                }
            
            # 使用EnhancedDataCleaner生成数据摘要
            datacle = EnhancedDataCleaner()
            summary = datacle.generate_data_summary(df)
            
            # 添加列的显示名称信息
            if "column_display_names" in dataset_data and dataset_data["column_display_names"]:
                summary["列显示名称"] = dataset_data["column_display_names"]
            
            logger.info(f"数据集 {dataset_id} 的数据摘要生成完成")
            return summary
            
        except Exception as e:
            logger.error(f"生成数据集摘要时出错: {str(e)}")
            return {
                "错误": f"生成数据摘要失败: {str(e)}"
            }

    @staticmethod
    async def create_dataset_fields(df, dataset_id, session):
        """
        根据DataFrame创建数据集字段记录
        
        参数:
            df: pandas DataFrame对象，包含数据集数据
            dataset_id: 数据集ID
            session: 数据库会话对象
            
        返回:
            创建的字段数量
        """
        from app.models.db import DataField
        import pandas as pd
        import re
        from loguru import logger
        
        logger.debug(f"为数据集 {dataset_id} 创建字段记录")
        
        # Python类型到数据库类型的映射
        TYPE_MAPPING = {
            'int64': 'integer',
            'int32': 'integer',
            'int16': 'integer',
            'int8': 'integer',
            'uint64': 'integer',
            'uint32': 'integer',
            'uint16': 'integer',
            'uint8': 'integer',
            'float64': 'float',
            'float32': 'float',
            'object': 'string',
            'datetime64[ns]': 'datetime',
            'bool': 'boolean',
            'category': 'string'
        }
        
        try:
            # 获取数据类型和统计信息以判断语义类型
            df_types = df.dtypes
            
            # 识别可能的类别字段（唯一值比例较低的字符串字段）
            categorical_cols = []
            for col in df.select_dtypes(include=['object']).columns:
                unique_ratio = df[col].nunique() / len(df) if len(df) > 0 else 1
                if unique_ratio < 0.5 and df[col].nunique() < 20:
                    categorical_cols.append(col)
            
            # 识别可能的日期字段
            date_cols = []
            for col in df.columns:
                # 如果已经是日期类型
                if pd.api.types.is_datetime64_dtype(df[col]):
                    date_cols.append(col)
                # 尝试检测字符串格式的日期
                elif df[col].dtype == 'object':
                    sample = df[col].dropna().iloc[:100] if len(df[col].dropna()) > 0 else pd.Series()
                    if any(re.match(r'\d{4}[-/]\d{1,2}[-/]\d{1,2}', str(x)) for x in sample) or \
                       any(re.match(r'\d{1,2}[-/]\d{1,2}[-/]\d{4}', str(x)) for x in sample):
                        date_cols.append(col)
            
            # 创建字段记录
            created_fields = 0
            for i, col_name in enumerate(df.columns):
                # 基本数据类型
                python_type = str(df_types[col_name])
                data_type = TYPE_MAPPING.get(python_type, 'string')
                
                # 确定语义类型
                if col_name in categorical_cols:
                    semantic_type = "分类"
                elif col_name in date_cols:
                    semantic_type = "日期"
                elif data_type in ['integer', 'float', 'decimal']:
                    semantic_type = "数值"
                elif pd.api.types.is_bool_dtype(df[col_name]) or data_type == 'boolean':
                    semantic_type = "布尔"
                elif 'id' in col_name.lower() or '编号' in col_name:
                    semantic_type = "标识符"
                elif any(kw in col_name.lower() for kw in ['phone', 'mobile', '电话', '手机']):
                    semantic_type = "电话"
                elif any(kw in col_name.lower() for kw in ['email', 'mail', '邮箱']):
                    semantic_type = "邮箱"
                else:
                    semantic_type = "文本"
                
                # 判断是否可为空
                is_nullable = df[col_name].isna().any()
                
                # 创建字段记录
                new_field = DataField(
                    dataset_id=dataset_id,
                    name=col_name,
                    display_name=col_name,  # 默认使用字段名作为显示名
                    description="",  # 默认为空，可由用户后续补充
                    data_type=data_type,
                    original_data_type=python_type,
                    is_primary_key=False,  # 默认非主键
                    is_nullable=is_nullable,
                    visible="1",  # 默认可见
                    semantic_type=semantic_type,
                    ordinal_position=i
                )
                
                session.add(new_field)
                created_fields += 1
            
            # 提交更改
            await session.flush()
            
            logger.info(f"成功为数据集 {dataset_id} 创建了 {created_fields} 个字段记录")
            return created_fields
            
        except Exception as e:
            logger.error(f"创建数据集字段记录失败: {str(e)}")
            raise

    @staticmethod
    def get_db_type_from_url(url: str) -> str:
        """根据数据库URL判断数据库类型"""
        if url.startswith('sqlite'):
            return 'sqlite'
        elif url.startswith('mysql'):
            return 'mysql'
        elif url.startswith('postgresql'):
            return 'postgresql'
        elif url.startswith('oracle') or url.startswith('cx_oracle'):
            return 'oracle'
        else:
            return 'unknown'

    @staticmethod
    def compress_json_array(json_array):
        """
        压缩JSON数组，将列名提取出来，减少token消耗
        
        输入格式：[{"产品名称": "手机", "产品类别": "移动设备", "总销售数量": 37.0}, ...]
        输出格式：{"columns": ["产品名称", "产品类别", "总销售数量"], "data": [["手机", "移动设备", 37.0], ...]}
        
        Args:
            json_array: 要压缩的JSON数组
            
        Returns:
            dict: 压缩后的数据结构
        """
        if not json_array or not isinstance(json_array, list):
            return {"columns": [], "data": []}
        
        # 如果数组为空，返回空结构
        if len(json_array) == 0:
            return {"columns": [], "data": []}
        
        # 获取列名（从第一个对象中提取）
        first_item = json_array[0]
        if not isinstance(first_item, dict):
            return {"columns": [], "data": json_array}
        
        columns = list(first_item.keys())
        
        # 提取数据
        data = []
        for item in json_array:
            if isinstance(item, dict):
                row = [item.get(col) for col in columns]
                data.append(row)
            else:
                # 如果遇到非字典类型，直接添加
                data.append(item)
        
        return {
            "columns": columns,
            "data": data
        }

    @staticmethod
    def decompress_json_array(compressed_data):
        """
        解压缩JSON数组，将压缩格式还原为原始格式
        
        输入格式：{"columns": ["产品名称", "产品类别", "总销售数量"], "data": [["手机", "移动设备", 37.0], ...]}
        输出格式：[{"产品名称": "手机", "产品类别": "移动设备", "总销售数量": 37.0}, ...]
        
        Args:
            compressed_data: 压缩后的数据结构
            
        Returns:
            list: 解压缩后的JSON数组
        """
        if not compressed_data or not isinstance(compressed_data, dict):
            return []
        
        columns = compressed_data.get("columns", [])
        data = compressed_data.get("data", [])
        
        # 如果没有列名或数据，返回空数组
        if not columns or not data:
            return []
        
        # 重建JSON数组
        json_array = []
        for row in data:
            if isinstance(row, list) and len(row) == len(columns):
                # 将行数据与列名配对
                item = dict(zip(columns, row))
                json_array.append(item)
            else:
                # 如果行数据不是列表或长度不匹配，跳过
                continue
        
        return json_array

    @staticmethod
    def calculate_compression_ratio(original_json_array, compressed_data):
        """
        计算压缩比例，用于衡量节省的token数量
        
        Args:
            original_json_array: 原始JSON数组
            compressed_data: 压缩后的数据结构
            
        Returns:
            dict: 包含压缩统计信息的字典
        """
        try:
            # 转换为JSON字符串来计算长度
            original_str = json.dumps(original_json_array, ensure_ascii=False)
            compressed_str = json.dumps(compressed_data, ensure_ascii=False)
            
            original_length = len(original_str)
            compressed_length = len(compressed_str)
            
            # 计算压缩比例
            if original_length > 0:
                compression_ratio = (original_length - compressed_length) / original_length * 100
                savings = original_length - compressed_length
            else:
                compression_ratio = 0
                savings = 0
            
            return {
                "original_length": original_length,
                "compressed_length": compressed_length,
                "savings": savings,
                "compression_ratio": round(compression_ratio, 2),
                "compression_ratio_desc": f"压缩了 {round(compression_ratio, 2)}%"
            }
        except Exception as e:
            return {
                "original_length": 0,
                "compressed_length": 0,
                "savings": 0,
                "compression_ratio": 0,
                "compression_ratio_desc": f"计算失败: {str(e)}"
            }

    @staticmethod
    def compress_json_array_smart(json_array, min_rows_for_compression=3):
        """
        智能压缩JSON数组，只有当数据量足够大时才进行压缩
        
        Args:
            json_array: 要压缩的JSON数组
            min_rows_for_compression: 进行压缩的最小行数阈值
            
        Returns:
            dict: 包含压缩结果和是否压缩的信息
        """
        if not json_array or not isinstance(json_array, list) or len(json_array) < min_rows_for_compression:
            return {
                "compressed": False,
                "data": json_array,
                "reason": f"数据量小于{min_rows_for_compression}行，无需压缩"
            }
        
        # 进行压缩
        compressed_data = DataHelp.compress_json_array(json_array)
        
        # 计算压缩效果
        compression_stats = DataHelp.calculate_compression_ratio(json_array, compressed_data)
        
        # 如果压缩效果不明显（节省不到10%），则不压缩
        if compression_stats["compression_ratio"] < 10:
            return {
                "compressed": False,
                "data": json_array,
                "reason": f"压缩效果不明显（仅节省{compression_stats['compression_ratio']}%），保持原格式"
            }
        
        return {
            "compressed": True,
            "data": compressed_data,
            "compression_stats": compression_stats
        }
